package fun.easycode.datastream;

import com.alibaba.ttl.TtlCallable;
import com.alibaba.ttl.TtlRunnable;
import lombok.Getter;

import java.util.concurrent.*;

/**
 * Quick全量线程池
 * 对于全量任务使用这个线程池进行处理
 *
 * @author xuzhen97
 */
public class DataCompleteExecutor {
    /**
     * 线程池
     */
    @Getter
    private final ThreadPoolExecutor executor;

    /**
     * 线程数量
     */
    private final static int THREAD_COUNT = 8;

    public DataCompleteExecutor() {
        // 线程池创建4个线程，任务队列容量2048，如果任务队列满了，则采用调用者线程执行
        // 线程池创建为4，在项目创建初期先分配2核4G的配置来测试稳定性
        this.executor = new ThreadPoolExecutor(THREAD_COUNT, THREAD_COUNT, 0
                , TimeUnit.SECONDS, new LinkedBlockingQueue<>(2048)
                , new ThreadPoolExecutor.CallerRunsPolicy());
    }

    /**
     * 提交任务
     *
     * @param callable 任务
     * @param <T>      任务执行结果类型
     * @return 任务执行结果
     */
    public <T> Future<T> submitTask(Callable<T> callable) {
        return executor.submit(TtlCallable.get(()->{
            try{
                return callable.call();
            }catch (Exception e){
                e.printStackTrace();
                throw e;
            }
        }));
    }

    /**
     * 提交任务
     *
     * @param runnable 任务
     */
    public void submitTask(Runnable runnable) {
        executor.submit(TtlRunnable.get(()->{
            try{
                runnable.run();
            }catch (Exception e){
                e.printStackTrace();
                throw e;
            }
        }));
    }
}
